package com.lazar.dw.dws

import java.util
import java.util.concurrent.TimeUnit

import com.alibaba.fastjson.{JSON, JSONArray, JSONObject}
import com.lazar.model.AdClick
import com.lazar.utils.SourceKafka
import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, Watermark, WatermarkGenerator, WatermarkGeneratorSupplier, WatermarkOutput, WatermarkStrategy}
import org.apache.flink.streaming.api.scala.{DataStream, KeyedStream, OutputTag, StreamExecutionEnvironment}
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.flink.api.scala._
import org.apache.flink.cep.functions.PatternProcessFunction
import org.apache.flink.cep.scala.{CEP, PatternStream}
import org.apache.flink.cep.scala.pattern.Pattern
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector

object BlackUserCEP {
  def main(args: Array[String]): Unit = {
      val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
      val kafkaConsumer: FlinkKafkaConsumer[String] = new SourceKafka().getKafkaSource("eventlog")
      val data: DataStream[String] = env.addSource(kafkaConsumer)
      /*
      area/uid/productId/timestamp
       */
      val adClickStream: DataStream[AdClick] = data.map(x => {
        val adJsonObject: JSONObject = JSON.parseObject(x)
        val attrObject: JSONObject = adJsonObject.getJSONObject("attr")
        val area: String = attrObject.get("area").toString
        val uid: String = attrObject.get("uid").toString
        var productId: String = null
        var timestamp: Long = 0L
        val array: JSONArray = adJsonObject.getJSONArray("lagou_event")
        array.forEach(x => {
          val nObject: JSONObject = JSON.parseObject(x.toString)
          if (nObject.get("name").equals("ad")) {
            val adObject: JSONObject = nObject.getJSONObject("json")
            productId = adObject.get("product_id").toString
            timestamp = TimeUnit.MICROSECONDS.toSeconds(nObject.get("time").toString.toLong)
          }
        })
        AdClick(area, uid, productId, timestamp)
      })

    val watermarkStream: DataStream[AdClick] = adClickStream.assignTimestampsAndWatermarks(new WatermarkStrategy[AdClick] {
      override def createWatermarkGenerator(context: WatermarkGeneratorSupplier.Context): WatermarkGenerator[AdClick] = {
        new WatermarkGenerator[AdClick] {
          var maxTimestamp = Long.MinValue
          override def onEvent(t: AdClick, l: Long, watermarkOutput: WatermarkOutput): Unit = {
            maxTimestamp = math.max(maxTimestamp,t.timestamp)
          }

          override def onPeriodicEmit(watermarkOutput: WatermarkOutput): Unit = {
            watermarkOutput.emitWatermark(new Watermark(maxTimestamp - 500L))
          }
        }
      }
    }.withTimestampAssigner(new SerializableTimestampAssigner[AdClick] {
      override def extractTimestamp(t: AdClick, l: Long): Long = {
        t.timestamp
      }
    }))

    val keyStream:KeyedStream[AdClick,(String,String)] = watermarkStream.keyBy(x => (x.uid, x.productId))
    val pattern: Pattern[AdClick,AdClick] = Pattern.begin[AdClick]("start").timesOrMore(10).within(Time.seconds(10))
    val patternStream: PatternStream[AdClick] = CEP.pattern(keyStream,pattern)

    patternStream.process(new PatternProcessFunction[AdClick,AdClick] {
      override def processMatch(map: util.Map[String, util.List[AdClick]], context: PatternProcessFunction.Context, collector: Collector[AdClick]): Unit = {
        System.out.println(map)
      }
    }).print()

    env.execute()
  }
}
